package org.example

import org.apache.spark.sql.SparkSession



object yy09 {
  def main(args: Array[String]): Unit = {
    //SPARK运行环境
    val spark = SparkSession.builder().master("local[*]").appName("spark").getOrCreate()
    val sc = spark.sparkContext

    //创建rdd 不指定分区就会打散
    val rdd1 = sc.makeRDD(List(1,5,3,7,6),1)
    val rdd2 = sc.parallelize(List(('a',75),('b',95),('c',56),('b',95)))
    //spak数据分析方法
    val rdd3 = sc.makeRDD(List(1,2,3,4,5),1)

    //工资
//    val first_half = sc.textFile("E:\\23spark209\\Employee_salary_first_half.csv")
//    val second_half = sc.textFile("E:\\23spark209\\Employee_salary_second_half.csv")
//    val drop_first = first_half.mapPartitionsWithIndex((ix, it) => {
//      if (ix == 0) it.drop(1)
//      it
//    })
//    val drop_second = second_half.mapPartitionsWithIndex((ix, it) => {
//      if (ix == 0) it.drop(1)
//      it
//    })
//    val split_first = drop_first.map(
//      Line => {
//        val data = Line.split(","); (data(1), data(6).toInt)
//        try {
//          (data(1),data(6).toInt)
//        }
//        catch {
//          case e:NumberFormatException =>(data(1),0)
//        }
//      })
//    val split_second = drop_second.map(
//      Line => {
//        val data = Line.split(","); (data(1), data(6).toInt)
//        try {
//          (data(1), data(6).toInt)
//        }
//        catch {
//          case e: NumberFormatException => (data(1), 0)
//        }
//      })


//    val filter_first = split_first.filter(x => x._2 > 200000).map(x => x._1)
//    val filter_second = split_second.filter(x => x._2 > 200000).map(x => x._1)
//    val name = filter_first.union(filter_second).distinct()
//    //合并并降重
//    name.collect().foreach(println)



    //交集
    //  rdd1.intersection(rdd3).foreach(println)
    //并集union
    //  rdd1.union(rdd3).foreach(println)
    //差集
    //  rdd1.subtract(rdd3).foreach(println)
    //zip拉链  分区数要一致 形成元组(key,value)
    //  rdd1.zip(rdd3).foreach(println)
    //笛卡尔积 cartesian 1跟3中的每个原始都会重新合成一组 5*5=25组
    //  rdd1.cartesian(rdd3).foreach(println)
    // fillter rdd2中筛选出大于90的数据
    //  rdd2.filter(tp => tp._2 >=90).foreach(println)
    // 简化
   // rdd2.filter(_._2 >= 90).foreach(println)
    //去重distinct
   // rdd2.distinct().foreach(println)

    //输出结果
    //  rdd1.foreach(println)
    //  rdd2.collect().foreach(println)

    val data1 =sc.makeRDD(List(('a',90),('b',78),('c',66),('d',78)),2)
    val data2 =sc.makeRDD(List(('a',5),('b',4),('c',3),('a',6)),2)
//    data1.reduceByKey((x,y) => x + y).foreach(println)
//    data1.groupByKey().foreach(println)
//    data1.groupBy(_._1).foreach(println)
    //join
//    data1.join(data2).foreach(println)
    data1.leftOuterJoin(data2).foreach(println)


    sc.stop()

  }

}
